package com.joven.interface2.conf;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;

import com.joven.core.model.Person;
import com.joven.interface2.mapper.PersonMapper;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;

import java.nio.charset.StandardCharsets;

/**
 * @author joven_wong
 * 2021/6/10 15:26
 */
@RocketMQTransactionListener(txProducerGroup = "yu")
public class LocalTransactionListener implements RocketMQLocalTransactionListener {

    @Autowired
    PersonMapper personMapper;

    /**
     * rocketmq 消息发送成功之后，提交本地事务
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
//        Person person = JSON.parseObject(new String((byte[])message.getPayload()), Person.class);
        Person person = JSON.parseObject(new String((byte[])message.getPayload(), StandardCharsets.UTF_8), Person.class);
        //执行本地事务
        int i= personMapper.insert(person);

//        Person person = JSON.parseObject(message.getHeaders().get("args",String.class),Person.class);
        Long arg = (Long) o;

        if(i>0){
            System.out.println(String.format("half message\npayload:%s, arg:%s, transactionId:%s", person, arg, message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID)));
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }

    /**
     * rocketmq 回查时，告诉它要提交，还是回滚
     * @param message
     * @return RocketMQLocalTransactionState
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        Person person = JSON.parseObject(new String((byte[])message.getPayload(), StandardCharsets.UTF_8), Person.class);
        //        Person person = JSON.parseObject(message.getHeaders().get("args",String.class),Person.class);
        QueryWrapper<Person> wrapper = new QueryWrapper<>(person);
        Person personOri = personMapper.selectOne(wrapper);
        if(personOri!=null){
            return RocketMQLocalTransactionState.COMMIT;
        }else{
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}
